package com.siteseat.crudeq.kernel.impl;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.Lock;

import com.siteseat.crudeq.kernel.Chunk;
import com.siteseat.crudeq.kernel.ChunkFileFilter;
import com.siteseat.crudeq.kernel.ChunkInfo;
import com.siteseat.crudeq.kernel.ChunkManager;
import com.siteseat.crudeq.kernel.KernelFactory;
import com.siteseat.crudeq.kernel.QueueProfile;
import com.siteseat.crudeq.kernel.exception.CreateChunkFailureException;

/**
 *
 */
public class PersistedChunkManager extends ChunkManager {
  
  /**
   * @param profile
   */
  public PersistedChunkManager( QueueProfile profile ) {
    super( profile );
  }
  
  /**
   * {@inheritDoc}
   */
  @Override
  protected void load() throws IOException, CreateChunkFailureException {
    QueueProfile prof = getProfile();

    ChunkFileFilter filter = getFactory().newChunkFileFilter( prof );
    new File( prof.getChunkFolderPath() ).listFiles( filter );

    ArrayList<ChunkInfo> all = new ArrayList<ChunkInfo>( filter.getResult().values() );
    Collections.sort( all );

    if( prof.isCheckChunk() ) {
      checkAllChunks();
    }

    Long currCid = Long.valueOf( filter.getMaxChunkId() + 1 );
    Chunk c = newChunkObject( currCid, getProfile().getChunkSizeByBytes() );

    setActiveChunk( c );
    all.add( c.getInfo() );
    
    setAllChunkInfo( all );

    prof.setCheckChunk( true );
  }


  /**
   * {@inheritDoc}
   */
  @Override
  protected Chunk newChunkObject( Long cid, int capacity ) throws IOException {
    KernelFactory f = getFactory();
    ChunkInfo ci = f.newChunkInfo( getProfile(), cid );
    return f.newChunk( ci, capacity );
  }

  /**
   * @return
   * @throws InterruptedException
   * @throws InterruptedException
   */
  protected LinkedHashMap<Long, Chunk> whichNeedToSave() throws InterruptedException {
    float capacity = ( 1/* this.onlined */+ getOfflinedChunks().size() ) / 0.75f;
    LinkedHashMap<Long, Chunk> r = new LinkedHashMap<Long, Chunk>( (int)capacity );

    Chunk c = getActiveChunk();
    if( c != null && c.notAllSaved() ) {
       r.put( c.getInfo().getChunkId(), c );
    }
    
    Lock flk = getOfflinedChunkLock().readLock();
    flk.lockInterruptibly();
    try {
      for( Map.Entry<Long, Chunk> ent: getOfflinedChunks().entrySet() ) {
        c = ent.getValue();
        if( c.notAllSaved() ) {
          r.put( ent.getKey(), c );
        }
      }
    } finally {
      flk.unlock();
    }

    return r;
  }

  /**
   * @param c
   */
  protected void saveSpecified( Chunk c ) {
    if( c.notAllSaved() ) {
      try {
        c.save();
      } catch( IOException e ) {
        log.error( "Failed to save chunk <" + c.getInfo().getChunkId() + "> due to; ", e );
      }
    }
  }

  /**
   * @param candidates
   * @throws InterruptedException
   */
  protected void saveSpecified( Set<Map.Entry<Long, Chunk>> candidates ) {
    for( Map.Entry<Long, Chunk> ent: candidates ) {
      Chunk c = ent.getValue();
      saveSpecified( c );
    }
  }

  /**
   * @param candidates
   * @throws InterruptedException
   * @throws IOException
   */
  protected void removeSavedOfflinedChunk( Set<Map.Entry<Long, Chunk>> candidates ) throws InterruptedException {

    Lock lk = getOfflinedChunkLock().writeLock();
    lk.lockInterruptibly();

    try {
      for( Map.Entry<Long, Chunk> ent: candidates ) {
        Chunk c = getOfflinedChunks().remove( ent.getKey() );
        if( c != null ) {
          closeSpecified( c );
        }
      }
    } finally {
      lk.unlock();
    }
  }

  /**
   * @param c
   */
  protected void closeSpecified( Chunk c ) {
    try {
      c.close();
    } catch( IOException e ) {
      log.error( "Error during close chunk <" + c.getInfo().getChunkId() + "> due to: ", e );
    }
  }

  /**
   * @throws InterruptedException
   * @throws IOException
   */
  public void save() throws InterruptedException {
    Set<Map.Entry<Long, Chunk>> candidates = whichNeedToSave().entrySet();

    if( candidates.size() > 0 ) {
      saveSpecified( candidates );
      removeSavedOfflinedChunk( candidates );
    }
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void doDestroy() throws Exception {
    boolean hasError = false;
    System.out.println("com.siteseat.crudeq.kernel.ChunkManager.doDestroy()");
    for( Chunk c: whichNeedToSave().values() ) {
      try {
        saveSpecified( c );
        closeSpecified( c );
      } catch( Exception e ) {
        hasError = true;
        log.error( "Error during saving chunk", e );
      }
    }

    getActiveChunk().close();
    
    getProfile().setCheckChunk( hasError );
  }

}
